package com.atguigu.flink.window;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import com.atguigu.flink.utils.MyUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * Created by Smexy on 2023/1/30
 * 
 *      分类的规则三种
 *          1.是否keyBy
 *          2.基于时间还是个数
 *                  个数: 处理函数中就有 GlobalWindow
 *                  时间: 处理函数中就有 TimeWindow
 *          3.特性(滚动|滑动|会话)
 *          
 *       不keyBy,基于时间的 滚动|滑动|会话
 *
 *    -----------
 *      不keyBy :  xxxAll
 *      keyBy:     xxx
 *
 *      时间:  window
 *      个数:  countWindow
 *
 *   -----------
 *      时间观
 *              flink中有两种时间
 *                   目前都以processingTime为案例！
 *                  处理时间 processingTime: 计算机的物理时钟。
 *                                          来了一条数据，如何知道当前几点?
 *                                              看程序的物理时钟！
 *                                              Event(ts=1点)
 *                                              计算机的时间: 2点
 *                                                  最终确定时间:2点
 *
 *                  需要集合watermark机制使用。
 *                  事件时间 EventTime: 从数据中抽取时间字段作为当前的时间。
 *                                          来了一条数据，如何知道当前几点?
 *                                              Event(ts=1点)
 *                                              计算机的时间: 3点
 *                                                  最终确定时间:1点
 */
public class Demo2_NoKeyByTimeWindow
{
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
                env
                   .socketTextStream("hadoop103", 8888)
                   .map(new WaterSensorMapFunction())
                   //基于处理时间的滚动窗口   size = slide = 5s
                   //.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                   //基于处理时间的滑动窗口   size = 5s,slide = 3s
                   //.windowAll(SlidingProcessingTimeWindows.of(Time.seconds(5),Time.seconds(3)))
                   //基于处理时间的会话窗口  gap（两条数据的时间间隔），只要两条数据的间隔 = gap，旧的窗口就关闭
                   .windowAll(ProcessingTimeSessionWindows.withGap(Time.seconds(5)))
                   .process(new ProcessAllWindowFunction<WaterSensor, String, TimeWindow>()
                   {
                       /*
                            TimeWindow： 标识当前是一个基于时间的窗口
                                    [start,end)
                        */
                       @Override
                       public void process(Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {

                            //获取当前时间窗口
                           TimeWindow window = context.window();
                           System.out.println("当前窗口范围:"+MyUtil.parseTime(window));

                           List<WaterSensor> list = MyUtil.parseList(elements);
                           out.collect(list.toString());

                       }
                   })
                   .print();

        
                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
        
    }
}
